package zhongche;

import com.zhongche.bean.Person;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.drools.core.io.impl.ClassPathResource;
import org.kie.api.KieBase;
import org.kie.api.runtime.StatelessKieSession;
import org.kie.internal.utils.KieHelper;

public class SimpleRulesTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.fromElements(new Person("1", 8), new Person("2", 10),
                new Person("3", 18),
                new Person("4", 15),
                new Person("5", 25))
                .map(new RichMapFunction<Person, Person>() {

                    private StatelessKieSession statelessKieSession;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        KieHelper kieHelper = new KieHelper();
                        kieHelper.kfs.write(new ClassPathResource("rules/simple.drl"));
                        KieBase kbase = kieHelper.build();
                        statelessKieSession = kbase.newStatelessKieSession();
                    }

                    @Override
                    public Person map(Person value) throws Exception {

                        statelessKieSession.execute(value);
                        return value;

                    }
                })
                .print("result");

        env.execute();

    }
}
